#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <sys/epoll.h>
#include <assert.h>
#include <arpa/inet.h>
#include <stdarg.h>
#include <signal.h>

#include <kvsp.h>
#include <kvs.h>
#include "server_func.h"

int deal_listen(int port)
{
    int listen_fd;
	int ret, sock_buf_size;

    struct sockaddr_in servaddr;

    if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
    {
        perror("open socket file error");
        return -1;
    }

	sock_buf_size=625*1024; /*BDP*/
ret=setsockopt(listen_fd, SOL_SOCKET, SO_SNDBUF, (char*)&sock_buf_size, sizeof(sock_buf_size));
	if (ret==-1)
   	 {
       	 perror("setsockopt error");
       	 return -1;
   	 }
		

    int flag = 1;

    if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, (const void *)&flag, sizeof(flag)) == -1)
setsockopt(listen_fd, SOL_SOCKET, SO_RCVBUF, (char*)&sock_buf_size, sizeof(sock_buf_size)); 
    //setnonblock(listen_fd);

    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(port);
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);

    if (bind(listen_fd, (struct sockaddr *)&servaddr, sizeof(servaddr)) < 0)
    {
        perror("bind error");
        return -1;
    }

    if (listen(listen_fd, BACKLOG) < 0)
    {
        perror("listen error");
        return -1;
    }

    return listen_fd;
}

int deal_accept(int listen_fd)
{
    int connect_fd;
    struct sockaddr_in client_addr;
    socklen_t addr_len;

    addr_len= sizeof(client_addr);

    if ((connect_fd = accept(listen_fd, (struct sockaddr*)&client_addr, &addr_len)) < 0)
    {
        perror("accept error");
        exit(-1);
    } 

    slog(stderr, "Client IP:%s Port:%d sockfd:%d\n", inet_ntoa(client_addr.sin_addr), client_addr.sin_port, connect_fd);
    
    setnonblock(connect_fd);
    return connect_fd;
}

int init(char* pathname, int init_type, char** kvs_image)
{
    KVS_SET kvs;
    close(2);                                       //close stderr
    freopen("server.log", "a+", stderr);     
    kvs.init_type=init_type;                       // or KVS_LOAD if load kvs.
    kvs.size_in_g=1000;
    strcpy(kvs.disk_file, pathname);
    strcpy(kvs.disk_log, "/tmp/kvsd_disk.log");
    strcpy(kvs.sync_log, "/tmp/kvsd_sync.log");
    strcpy(kvs.index_log, "/tmp/kvsd_index.log");
/*FIXME: if kvs.disk_file is invalid, there is no warninig! */
    kv_init(&kvs, kvs_image);

    return 0;
}

int com_parse(EHPTR ehptr)
{
    int cfd = ehptr->connect_fd;
    COMPACK *cp = (COMPACK *)ehptr->command;
    int n;

    n = read(cfd, (void *)cp, sizeof(COMPACK));
    
    if (n==-1&&(errno==EAGAIN||errno==EWOULDBLOCK))
        return -1;
    if (n == 0)
    {
        epoll_del(ehptr);
        return -1;
    }
   
    if(n!=16)
    {
        printf("Line 249:n=%d!\n",n);
        exit(-1);
    }
    
    if (cp->type==PUT)
    {
        slog(stderr, "Command:PUT\n");
        ehptr->operation=PUT;
        ehptr->buffsize=cp->key_length+cp->value_length;
    }
    else if (cp->type==GET)
    {
        slog(stderr, "Command:GET\n");
        ehptr->operation=GET;
        ehptr->buffsize=cp->key_length;
    }
    else if (cp->type==DELETE)
    {
        slog(stderr, "Command:DELETE\n");
        ehptr->operation=DELETE;
        ehptr->buffsize=cp->key_length;
    }
    ehptr->buffer=(char *)malloc(ehptr->buffsize);

    assert(ehptr->buffer);
    
    memset(ehptr->buffer, 0, ehptr->buffsize);
    ehptr->state=DATA_RECV;

    return 0;
}

int data_recv(EHPTR ehptr)
{
    char filename[64];
    int cfd = ehptr->connect_fd;
    int efd = ehptr->epoll_fd;
    int key_len, val_len, n, valget_len, toread;
    char *value_get;
    struct epoll_event ev;
    COMPACK *cp=(COMPACK *)ehptr->command;
    toread = ehptr->buffsize;
    n = read(cfd, ehptr->buffer+ehptr->buffpos, toread);
    
    if (n==-1&&(errno==EAGAIN||errno==EWOULDBLOCK))
        return -1;
    
    if (n ==0)
    {
        epoll_del(ehptr);
        return -1;
    }
    
    ehptr->buffpos += n;
    if (ehptr->buffpos>=toread)
    {   
        slog(stderr, "read %d bytes from cfd-%d\n", ehptr->buffpos, cfd);
        
        if (ehptr->operation==GET) 
        {
            if (kv_get_2(ehptr->buffer,toread,&value_get,&valget_len)!= 0)
            {

                free(ehptr->buffer);
                ehptr->buffer=NULL;
                ehptr->buffsize=0;
                ehptr->buffpos=0;
                slog(stderr, "GET failed!\n");
                cp->type=-1;
                cp->value_length=0;
            }
            else
            {                
                memset(filename, 0, sizeof(filename));
                strncpy(filename, ehptr->buffer, toread);
/*XXX
                printf("%s\n", filename);
                fp = fopen(filename, "wb");
                fwrite(value_get, sizeof(char), valget_len, fp);
                fclose(fp);
*/
                free(ehptr->buffer);
                ehptr->buffer = value_get;
                ehptr->buffsize=valget_len;
                ehptr->buffpos=0;
                slog(stderr, "GET succeed!\n");
                cp->type=0;
                cp->value_length=ehptr->buffsize;
            }
        }
        else if(ehptr->operation==PUT)
        {
            key_len=cp->key_length;
            val_len=cp->value_length;

            if (kv_put(ehptr->buffer,key_len,(ehptr->buffer)+key_len,val_len)==0)
            {
                slog(stderr, "PUT SUCCEED!\n");
                cp->type=0;
            }
            else 
            {
                slog(stderr, "PUT FAILED!\n");
                cp->type=-1;
            }
            cp->value_length=0;
            
            free(ehptr->buffer);
            ehptr->buffer=NULL;
            ehptr->buffsize=0;
            ehptr->buffpos=0;
        }
        else if (ehptr->operation==DELETE)
        {
            if (kv_delete(ehptr->buffer,ehptr->buffsize) == 0)
            {
                slog(stderr, "DELETE SUCCEED!\n");
                cp->type=0;
            }
            else 
            {
                slog(stderr, "DELETE FAILED!\n");
                cp->type=-1;
            }
            cp->value_length=0;
            
            free(ehptr->buffer);
            ehptr->buffsize=0;
            ehptr->buffpos=0;
       }

        ehptr->state=REPLY_SEND;
        ev.data.ptr=(void *)ehptr;
        ev.events=EPOLLOUT;
        epoll_ctl(efd, EPOLL_CTL_MOD, cfd, &ev);
    }

    return 0;
}

int reply_send(EHPTR ehptr)
{
    assert(ehptr!=NULL);
    int cfd, efd, n;
    COMPACK *cp;
    struct epoll_event ev;
    cfd = ehptr->connect_fd;
    efd = ehptr->epoll_fd;
    cp = (COMPACK *)ehptr->command;
    n = write(cfd, cp, sizeof(COMPACK));

    if (n==-1)
        return -1;
    if (n==0)
    {
        epoll_del(ehptr);
        return -1;
    }
    
    if(n!=16)
    {
        printf("Line 386:n!=16\n");
        exit(-1);
    }
    slog(stderr, "OPCODE-%d value_length-%d\n", cp->type, cp->value_length);
    if (ehptr->operation==GET&&cp->type==0)
    {
        ehptr->state=DATA_SEND;
    }
    else 
    {
        ehptr->state=COMMAND_RECV;
        ev.data.ptr=(void *)ehptr;
        ev.events=EPOLLIN;
        epoll_ctl(efd, EPOLL_CTL_MOD, cfd, &ev);
    }
    return 0;
}

int data_send(EHPTR ehptr)
{
    assert(ehptr);
    int cfd = ehptr->connect_fd;
    int efd = ehptr->epoll_fd;
    int n, towrite;
    struct epoll_event ev;

    towrite = ehptr->buffsize;
    n = write(cfd, ehptr->buffer+ehptr->buffpos, towrite-ehptr->buffpos);
    if (n==-1&&(errno==EAGAIN||errno==EWOULDBLOCK))
        return -1;
    if (n==0)
    {
        epoll_del(ehptr);
        return -1;
    }
    
    ehptr->buffpos+=n;
    if (ehptr->buffpos>=towrite)
    {
        slog(stderr, "Write %d bytes to cfd-%d\n", ehptr->buffpos, cfd);
        free(ehptr->buffer);
        ehptr->buffer=NULL;
        ehptr->buffpos=0;
        ehptr->buffsize=0;
        ehptr->state=COMMAND_RECV;
        ev.data.ptr=(void *)ehptr;
        ev.events=EPOLLIN;
        assert(epoll_ctl(efd, EPOLL_CTL_MOD, cfd, &ev)!=-1);
    }

    return 0;
}

void setnonblock(int fd)
{
    int opt = fcntl(fd, F_GETFL, 0);

    if (fcntl(fd, F_SETFL, opt|O_NONBLOCK) == -1)
    {
        perror("fcntl set failed");
        exit(-1);
    }
}

void init_event_handler(EHPTR ehptr, int cfd, int efd)
{
    assert(ehptr != NULL);

    ehptr->connect_fd = cfd;
    ehptr->epoll_fd = efd;
    memset(ehptr->command, 0, sizeof(ehptr->command));
    ehptr->buffer=NULL;
    ehptr->buffpos=0;
    ehptr->buffsize=0;
    ehptr->state=COMMAND_RECV;
}

int slog(FILE* fp, const char *format, ...)
{
    va_list _va_list;
    int result;
    if (format == NULL || fp == NULL) 
		return -1; 
    
    va_start(_va_list, format);
    result=vfprintf(fp, format, _va_list);
    va_end(_va_list);
    return result;
}

int epoll_add(EHPTR ehptr, uint32_t events)
{
    if (ehptr==NULL||events==0)
        return -1;

    struct epoll_event ev;
    ev.data.ptr = (void *)ehptr;
    ev.events = events;
    if (epoll_ctl(ehptr->epoll_fd, EPOLL_CTL_ADD, ehptr->connect_fd, &ev)==-1)
    {
        perror("epoll_add error");
        return -1;
    }
    return 0;
}

int epoll_del(EHPTR ehptr)
{
    assert(ehptr);
    if (ehptr->buffer)
        free(ehptr->buffer);
    epoll_ctl(ehptr->epoll_fd, EPOLL_CTL_DEL, ehptr->connect_fd, NULL);
    close(ehptr->connect_fd);
    free(ehptr);
    return 0;
}
